/**
 * Copyright (C), 2020-2020, 软件公司
 * FileName: ProducerTest1
 * Author:   cakin
 * Date:     2020/5/7
 * Description: 设置mandatory标志，且exchange未绑定队列
 */
package com.rabbitmq.msgdirection.mandatory;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.client.ReturnListener;
import com.rabbitmq.client.AMQP;

import com.rabbitmq.util.ConnectionUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import static com.rabbitmq.common.Constant.NUM3;

/**
 * @ClassName: ProducerTest
 * @Description: 设置mandatory标志，且exchange未绑定队列
 * @Date: 2020/5/7
 * @Author: cakin
 */
@Slf4j
public class ProducerTest1 {
    public static void main( String[] args ) {
        String exchangeName = "confirmExchange";
        String routingKey = "confirmRoutingKey";
        int count = NUM3;

        // 获取连接
        try {
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
            // 创建exchange
            channel.exchangeDeclare(exchangeName, "direct", true, false, null);
            // 发送持久化消息
            for (int i = 0; i < count; i++) {
                /**
                 *  exchangeName:默认情况下代理服务器端是存在一个""名字的exchange的,
                 *  因此如果不创建exchange的话我们可以直接将该参数设置成"",
                 *  如果创建了exchange的话我们需要将该参数设置成创建的exchange的名字
                 *  routingKey:路由键
                 *  mandatory:true 表示如果消息没有正确路由，消息会返回生产者
                 *  mandatory:false 表示如果消息没有正确路由，消息会丢失
                 */
                // 设置成了true，表示开启了mandatory标志，我们没有为当前exchange绑定任何队列
                 channel.basicPublish(exchangeName, routingKey, false,
                           MessageProperties.PERSISTENT_BASIC, ("第" + (i + 1) + "条消息").getBytes());
                // 设置成了false，表示关闭了mandatory标志，我们没有为当前exchange绑定任何队列
//                channel.basicPublish(exchangeName, routingKey, false,
//                        MessageProperties.PERSISTENT_BASIC, ("第" + (i + 1) + "条消息").getBytes());
            }
            channel.addReturnListener(new ReturnListener() {
                /**
                 * broker(代理服务器)发出basic.return方法之后，就会回调handleReturn方法，
                 * 在这个方法里面我们就可以进行消息的重新发布操作
                 * @param replyCode 312
                 * @param replyText NO_ROUTE：表示消息并没有路由到合适的队列中
                 * @param exchange 交换器
                 * @param routingKey 路由键
                 * @param properties 相关属性
                 * @param body 消息体
                 * @throws IOException
                 */
                @Override
                public void handleReturn( int replyCode, String replyText, String exchange, String routingKey,
                                          AMQP.BasicProperties properties, byte[] body ) throws IOException {
                    String message = new String(body);
                    log.info("replyCode:  " + replyCode);
                    log.info("replyText:  " + replyText);
                    log.info("Basic.Return返回的结果:  " + message);
                }
            });
        } catch (IOException e) {
            log.info("catch IOException");
        } catch (TimeoutException e) {
            log.info("catch TimeoutException");
        }
    }
}

